package Source

import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._

import scala.util.Random

object SourceFunctionTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val stream = env.addSource(new MySensorSource())

    stream.print()
    env.execute()
  }
}

/**
 * 自定义Source类
 */
class MySensorSource() extends SourceFunction[SensorReading] {
  //定义一个标志位flag，依赖表示数据源是否正常运行发出数据
  var flag = true

  override def run(sourceContext: SourceFunction.SourceContext[SensorReading]): Unit = {
    //定义一个随机数发生器
    val rand = new Random()

    //随机生成十个初始温度:(id:temp)
    var curTemp = 1.to(10).map(i => ("sensor_" + 1, rand.nextDouble() * 100))

    //定义无限循环，不停的产生数据，出发呗cancel
    while (flag) {
      //在上次温度的基础上更新温度值
      curTemp = curTemp.map(
        data => (data._1, data._2 + rand.nextGaussian())
      )
      //获取当前时间戳,将时间戳撞倒sensor，然后使用context发出数据
      val curTime = System.currentTimeMillis()
      curTemp.foreach(
        data => sourceContext.collect(SensorReading(data._1, curTime, data._2))
      )
      //.间隔一段时间
      Thread.sleep(1000)
    }
  }

  override def cancel(): Unit = flag = false
}